package cn.mesmile.flink.ecp;

import cn.hutool.core.date.DateUtil;
import com.mysql.cj.util.TimeUtil;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.util.List;
import java.util.Map;

/**
 * @author zb
 * @date 2021/8/28 20:48
 * @Description
 */
public class FlinkLoginCheckApp {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        environment.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

        DataStreamSource<String> ds = environment.socketTextStream("127.0.0.1", 8888);
        // username,time,code
        DataStream<Tuple3<String, String, Integer>> flatMap = ds.flatMap(new FlatMapFunction<String, Tuple3<String, String, Integer>>() {
            @Override
            public void flatMap(String value, Collector<Tuple3<String, String, Integer>> out) throws Exception {
                String[] split = value.split(",");
                out.collect(Tuple3.of(split[0],split[1],Integer.parseInt(split[2])));
            }
        });

        SingleOutputStreamOperator<Tuple3<String, String, Integer>> watermakerDS = flatMap
                .assignTimestampsAndWatermarks(WatermarkStrategy
                //延迟策略去掉了延迟时间,【时间是单调递增】，event中的时间戳充当了水印，
                .<Tuple3<String, String, Integer>>forMonotonousTimestamps()
                //生成一个延迟3s的固定水印,乱序的
                //.<Tuple3<String, String, Integer>>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(
                        (event, timestamp) -> {
                            //指定POJO的事件时间列
                            return DateUtil.parseDateTime(event.f1).getTime();
                        }
                ));
        // 分组
        KeyedStream<Tuple3<String, String, Integer>, String> keyedStream = watermakerDS
                .keyBy(new KeySelector<Tuple3<String, String, Integer>, String>() {
            @Override
            public String getKey(Tuple3<String, String, Integer> value) throws Exception {
                return value.f0;
            }
        });
        //定义模式
        Pattern<Tuple3<String, String, Integer>, Tuple3<String, String, Integer>> pattern = Pattern
                .<Tuple3<String, String, Integer>>begin("firstTimeLogin")
                .where(new SimpleCondition<Tuple3<String, String, Integer>>() {
            @Override
            public boolean filter(Tuple3<String, String, Integer> value) throws Exception {
                // -1 是登录失败错误码
                return value.f2 == -1;
            }
        })//.times(2).within(Time.seconds(10));//不是严格近邻
                .next("secondTimeLogin")
                .where(new SimpleCondition<Tuple3<String, String, Integer>>() {
                    @Override
                    public boolean filter(Tuple3<String, String, Integer> value) throws Exception {
                        return value.f2 == -1;
                    }
                }).within(Time.seconds(5));

        // 匹配检查
        PatternStream<Tuple3<String, String, Integer>> patternStream = CEP.pattern(keyedStream, pattern);
        // 获取结果
        SingleOutputStreamOperator<Tuple3<String, String, String>> select = patternStream.select(new PatternSelectFunction<Tuple3<String, String, Integer>, Tuple3<String, String, String>>() {
            // 账号，时间，时间
            @Override
            public Tuple3<String, String, String> select(Map<String, List<Tuple3<String, String, Integer>>> map) throws Exception {
                Tuple3<String, String, Integer> firstLoginFail = map.get("firstTimeLogin").get(0);
                Tuple3<String, String, Integer> secondLoginFail = map.get("secondTimeLogin").get(0);
                return Tuple3.of(firstLoginFail.f0, firstLoginFail.f1, secondLoginFail.f1);
            }
        });
        select.print("check_result:");

        environment.execute("cep job");
    }
}
